package com.bff.gaia.mix.examples.xjoinV4;
/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import com.bff.gaia.api.common.functions.MapFunction;
import com.bff.gaia.api.common.typeinfo.BasicTypeInfo;
import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.mix.api.TimeCharacteristic;
import com.bff.gaia.mix.api.datastream.DataMixStream;
import com.bff.gaia.mix.api.environment.MixStreamExecutionEnvironment;
import com.bff.gaia.mix.api.xjoin2.client.rdb.table.RdbSideTableInfo;
import com.bff.gaia.mix.api.xjoin2.client.rdb.table.RdbSideTableInfoBuilders;
import com.bff.gaia.mix.api.xjoin2.core.enums.AsyncXjoinCacheType;
import com.bff.gaia.mix.api.xjoin2.core.enums.XjoinMethod;
import com.bff.gaia.mix.api.xjoin2.datastream.XjoinMixStreamV4;
import com.bff.gaia.mix.examples.xjoin.benchmarkDataSource;
import com.bff.gaia.types.Row;

/**
 * Description: gaia-tiger
 * Created by tiger on 2019/4/15 16:56.
 * xjoin.example.mapBenchmark
 */
public class AsyncTest {
	public static void main(String[] args) throws Exception {
		MixStreamExecutionEnvironment env = MixStreamExecutionEnvironment.getExecutionEnvironment();
		env.setMixStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

//		final String jdbcDriver = "com.mysql.cj.jdbc.Driver";
//		final String jdbcUrl = "jdbc:mysql://192.168.100.203:3306/employees?characterEncoding=UTF-8&useSSL=false&serverTimezone=Hongkong&rewriteBatchedStatements=true&allowPublicKeyRetrieval=true";
//		final String jdbcUser = "root";
//		final String jdbcPassword = "123456";

//		final String jdbcDriver = "com.mysql.cj.jdbc.Driver";
//		final String jdbcUrl = "jdbc:mysql://127.0.0.1:3306/employees?characterEncoding=UTF-8&useSSL=false&serverTimezone=Hongkong&rewriteBatchedStatements=true&allowPublicKeyRetrieval=true";
//		final String jdbcUser = "gaia";
//		final String jdbcPassword = "labmysqlgaia";


		final String jdbcDriver = "com.mysql.cj.jdbc.Driver";
		final String jdbcUrl = "jdbc:mysql://127.0.0.1:3306/employees?characterEncoding=UTF-8&useSSL=false&serverTimezone=Hongkong&rewriteBatchedStatements=true&allowPublicKeyRetrieval=true";
		final String jdbcUser = "tiger";
		final String jdbcPassword = "tiger19960724";
//		生成员工编号（10000-11000） 构建数据流
//		List<Integer> ints = new ArrayList<>();
//		Random random = new Random();
//		int num=1000;
//		if(args.length==1){
//			num=Integer.valueOf(args[0]);
//		}
//		for (int i = 0; i < num; i++) {
//			ints.add(random.nextInt(999) + 10001);
//		}
//		DataStream<Integer> empNoStream =env.fromCollection(ints);
//
		RdbSideTableInfo dept_empTable = RdbSideTableInfoBuilders.buildMysqlTableInfo()
			.setTableName("dept_emp")
			.setUrl(jdbcUrl)
			.setUserName(jdbcUser)
			.setPassword(jdbcPassword)
			.addfieldInfos(new String[]{"emp_no int", "dept_no string", "from_date date", "to_date date"})
			.finish();

		RdbSideTableInfo deptTable = RdbSideTableInfoBuilders.buildMysqlTableInfo()
			.setTableName("departments")
			.setUrl(jdbcUrl)
			.setUserName(jdbcUser)
			.setPassword(jdbcPassword)
			.addfieldInfos(new String[]{"dept_no string", "dept_name string"})
			.finish();

		RdbSideTableInfo empTable =RdbSideTableInfoBuilders.buildMysqlTableInfo()
			.setTableName("employees")
			.setUrl(jdbcUrl)
			.setUserName(jdbcUser)
			.setPassword(jdbcPassword)
			.addfieldInfos(new String[]{
				"emp_no int",
				"birth_date date",
				"first_name string",
				"last_name string",
				"gender string",
				"hire_date date"})
			.finish();


		int num = 100;
		if (args.length == 1) {
			num = Integer.valueOf(args[0]);
		}
		DataMixStream<Integer> empNoStream = env.addSource(new benchmarkDataSource.SimpleSource(num));

		RowTypeInfo emp = new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO);
		DataMixStream<Row> empNoRowStream = empNoStream.map(new MapFunction<Integer, Row>() {
			@Override
			public Row map(Integer value) throws Exception {
				Row row = new Row(1);
				row.setField(0, value);

				return row;
			}
		}).returns(emp);

		DataMixStream<Row> dept_empStream=empNoRowStream.xjoinV4(dept_empTable)
			.projectStreamTableKey(0)
			.ProjectSideTableKey(0)
			.projectsideTableField(0,1)
			.AsyncXjoin()
			.setParallelism(3)
			.setCacheTimeout(10*1000)
			.setAsyncXjoinCacheType(AsyncXjoinCacheType.LRU)
			.apply();

		DataMixStream<Row> empExStream=dept_empStream.xjoinV4(empTable)
			.projectStreamTableKey(0)
			.ProjectSideTableKey(0)
			.projectsideTableField(0,2,3)
			.AsyncXjoin()
			.setParallelism(3)
			.setCacheTimeout(10*1000)
			.setAsyncXjoinCacheType(AsyncXjoinCacheType.LRU)
			.apply();

		DataMixStream<Row> empExAndDeptExStream=dept_empStream.xjoinV4(deptTable)
			.projectStreamTableKey(1)
			.ProjectSideTableKey(0)
			.AsyncXjoin()
			.setParallelism(3)
			.setCacheTimeout(10*1000)
			.setAsyncXjoinCacheType(AsyncXjoinCacheType.LRU)
			.apply();
		empExAndDeptExStream.print();
		env.execute();
	}
}